// Copyright (c) 2023 Huawei Device Co., Ltd.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::error::Error;
use crate::event_handler::event_queue::EventQueue;
use crate::event_handler::inner_event::DefaultType;
use crate::event_handler::{EventHandler, InnerEvent, Priority};
use crate::{Dispatcher, DispatcherHandle};
use std::cell::RefCell;
use std::future::Future;
use std::sync::{Arc, Weak};
use tokio::task::JoinHandle;

thread_local! {
    static RUNNER: RefCell<Option<EventRunnerHandle>> = RefCell::new(None);
}

/// An `EventRunner` binds with a single thread, and all the `EventHandler` starts on it will run on
/// the same thread only.
pub struct EventRunner {
    dispatcher: Dispatcher,
    tx: Arc<EventQueue<DefaultType>>,
}

pub(crate) async fn inner_loop(rx: Arc<EventQueue<DefaultType>>, runner: EventRunnerHandle) {
    // inner_loop runs in a current thread runtime, and the thread won't change across the await
    // point, so setting the thread local here is OK.
    RUNNER.with(|r| *r.borrow_mut() = Some(runner.clone()));
    while let Some(event) = rx.get().await {
        if let Some(strong) = event.owner() {
            let handler = EventHandler::new(strong, runner.clone());
            handler.distribute(event);
        }
    }
    RUNNER.with(|r| *r.borrow_mut() = None);
}

impl EventRunner {
    /// Creates a new `EventRunner`. User should call `run()` to make it start to work.
    /// See [`EventRunner::run`].
    pub fn create() -> Result<Self, Error> {
        let dispatcher = Dispatcher::new()?;
        let queue = Arc::new(EventQueue::new());
        let handler = EventRunnerHandle {
            dispatcher: dispatcher.handle(),
            tx: Arc::downgrade(&queue),
        };
        dispatcher.spawn(inner_loop(queue.clone(), handler));
        Ok(EventRunner {
            dispatcher,
            tx: queue,
        })
    }

    /// Creates a new `EventRunner`. A new thread will be created, and the `EventRunner` will run on
    /// it automatically.
    pub fn create_detached(thread_name: &str) -> Result<EventRunnerHandle, Error> {
        let dispatcher = Dispatcher::new_detached(thread_name)?;
        let queue = Arc::new(EventQueue::new());
        let runner = EventRunnerHandle {
            dispatcher: dispatcher.clone(),
            tx: Arc::downgrade(&queue),
        };
        dispatcher.spawn(inner_loop(queue, runner.clone()));
        Ok(runner)
    }

    /// Returns a handle to the `EventRunner`. The returned handle can be used to start
    /// [`EventHandler`] on this runner, to stop the runner, and can be cloned to allow moving
    /// the `EventRunnerHandle` to other threads.
    /// User should call this method before calling [`EventRunner::run`].
    ///
    /// [`EventHandler`]: crate::event_handler::EventHandler
    pub fn handle(&self) -> EventRunnerHandle {
        EventRunnerHandle {
            dispatcher: self.dispatcher.handle(),
            tx: Arc::downgrade(&self.tx),
        }
    }

    /// Runs the `EventRunner`. This function blocks the current thread, and will never
    /// return until [`EventRunnerHandle::stop`] is called.
    ///
    /// User should call [`EventRunner::handle`] before calling this method.
    ///
    /// # Panics
    ///
    /// This function panics if called within an asynchronous execution context.
    pub fn run(self) {
        self.dispatcher.run()
    }

    /// Gets the `EventRunner` on current thread.
    pub fn current() -> Option<EventRunnerHandle> {
        RUNNER.with(|r| r.borrow().clone())
    }
}

/// Handle to the [`EventRunner`]. The Handle can be freely cloned.
#[derive(Clone)]
pub struct EventRunnerHandle {
    dispatcher: DispatcherHandle,
    tx: Weak<EventQueue<DefaultType>>,
}

impl EventRunnerHandle {
    /// Stops the `EventRunner`. All the event handlers bind with it will be stopped too, and all
    /// the events haven't been processed will be dropped.
    pub fn stop(self) {
        if let Some(tx) = self.tx.upgrade() {
            tx.finish()
        }
        self.dispatcher.stop()
    }

    /// Spawns a task to the `EventRunner`.
    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
        F::Output: Send + 'static,
    {
        self.dispatcher.spawn(future)
    }

    pub(crate) fn put(
        &self,
        event: InnerEvent<DefaultType>,
        priority: Priority,
    ) -> Result<(), Error> {
        if let Some(tx) = self.tx.upgrade() {
            tx.put(event, priority);
            return Ok(());
        }
        Err(Error::Closed)
    }
}

#[cfg(test)]
mod ut_runner {
    use crate::event_handler::EventRunner;
    use std::thread;
    use tokio::runtime::Runtime;

    /// UT test for `EventRunner::create`.
    ///
    /// # Brief
    /// 1. Create a runner using `EventRunner::create`.
    /// 2. Get a handle of the runner.
    /// 3. Run the runner on a new thread.
    /// 4. Stop the runner and wait for the new thread exit.
    #[test]
    fn ut_create() {
        let runner = EventRunner::create().unwrap();
        let handle = runner.handle();
        let join = thread::spawn(|| runner.run());
        handle.stop();
        join.join().unwrap();
    }

    /// UT test for `EventRunnerHandle::spawn`.
    ///
    /// # Brief
    /// 1. Create a runner using `EventRunner::create_detached`.
    /// 2. Spawn a async task on the runner.
    /// 3. Wait for the result, and check the result.
    /// 4. Stop the runner.
    #[test]
    fn ut_spawn() {
        let handle = EventRunner::create_detached("test").unwrap();
        let runtime = Runtime::new().unwrap();
        let mut data = 10;
        let join = handle.spawn(async move {
            data += 1;
            data
        });
        let ret = runtime.block_on(join).unwrap();
        assert_eq!(ret, 11);
        handle.stop();
    }
}
